package ssssscala.JSONDataSource


import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, Row, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Auther: Mengkunxuan
  * Date:2018 /9/2814:16
  * Description:
  */
object JSONDataSource1 {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("JSONDataSource").setMaster("local")
    val sc = new SparkContext(conf)
    var sqlContext = new SQLContext(sc)
    //创建学生成绩DataFrame
    val studentScoresDF =sqlContext.read.json("C:\\Users\\Administrator\\Desktop\\第一阶段代码\\第80讲-Spark SQL：JSON数据源复杂综合案例实战\\文档\\students.json")
    //查询出分数大于80分的学生成绩信息,以及学生姓名
    studentScoresDF.registerTempTable("student_scores")
    val goodStudentScoresDF = sqlContext.sql("select name,score from student_scores where score >=80")
    val goodStudentNames = goodStudentScoresDF.rdd.map{row =>row(0)}.collect()
    //创建xue学生基本信息DataFrame
    val studentinfoJSONs = Array("{\"name\":\"Leo\",\"age\":18}",
    "{\"name\":\"Marry\",\"age\":17}",
      "{\"name\":\"Jack\",\"age\":19}"
    )
    val studentinfoJSONsRDD = sc.parallelize(studentinfoJSONs,3)
    val studentInfosDF = sqlContext.read.json(studentinfoJSONsRDD)
    //查询分数大于80分的学生的基本信息
    studentInfosDF.registerTempTable("student_infos")
    var sql = "select name,age from student_infos where name in("
    for(i <- 0 until goodStudentNames.length){
      sql += "'" + goodStudentNames(i)+"'"
      if (i< goodStudentNames.length -1){
        sql += ","
      }
    }
    sql += ")"
    val goodStudentInfosDF =sqlContext.sql(sql)
    //将分数大于80的学生成绩信息与基本信息进行join
    val goodStudentRDD =
      goodStudentScoresDF.rdd.map{ row =>(row.getAs[String]("name"),
      row.getAs[Long]("score"))}
        val goodStudentRDD2 = goodStudentInfosDF.rdd.map(
          row =>(row.getAs[String]("name"),row.getAs[Long]("age"))
        )

    val goodStudentRDD3 = goodStudentRDD.join(goodStudentRDD2);
    //将rdd转化为dataframe
    val goodStudentRowsRDD = goodStudentRDD3.map(info => Row(info._1,info._2._1.toInt,info._2._2.toInt))

    val structType = StructType(Array(
      StructField("name", StringType, true),
      StructField("score", IntegerType, true),
      StructField("age", IntegerType, true)))
    val goodStudentsDF:DataFrame = sqlContext.createDataFrame(goodStudentRowsRDD, structType)
    //将dataframe中的数据保存到json中
    goodStudentsDF.write.format("json").save("c:/shuchu/jsondatasourceshuchu")

  }

}
